# -*- coding: utf-8 -*-
from datetime import timedelta
from jms.aigroup import jms_ai_dw__yl_check_threesegcode_sample
from time import time
from utils.operators.rich_sql_sensor import RichSqlSensor

label = f"ml_check_threesegcode_sample_day_{int(time())}"
timeout = timedelta(minutes=30).seconds
doris_ai_group__ml_check_threesegcode_sample_day = RichSqlSensor(
    task_id='doris_ai_group__ml_check_threesegcode_sample_day',
    email=['yushuo@jtexpress.com','yl_bigdata@yl-scm.com'],
    pool='broker_load_pool',
    conn_id='doris_real_time',
    pre_sql=f"""
             TRUNCATE TABLE ai_group.ml_check_threesegcode_sample_day PARTITION (p{{{{ execution_date |  cst_ds_nodash }}}});
             LOAD LABEL ai_group.{label} (
                 DATA INFILE("hdfs://yl-bg-hdp/dw/hive/ai_dw.db/external/yl_check_threesegcode_sample_day/*") INTO TABLE ml_check_threesegcode_sample_day FORMAT AS 'PARQUET'
                 (sign_date,waybill_no,send_receiver_id,receiver_id,sender_province_name,sender_city_name,sender_area_name,receiver_province_name,receiver_city_name,receiver_area_name,sign_provider_name,sign_city_name,sign_area_name,receiver_detailed_address,pdd_threesegcode,sign_threesegcode,sign_network_code,pick_network_code,center_code)
                 SET (dt = DATE('{{{{ execution_date | cst_ds }}}}'))
              )
             WITH BROKER '{{{{ var.json.doris_real_time_brokers | random_choice }}}}'
             PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM ai_group WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM ai_group WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120), )

doris_ai_group__ml_check_threesegcode_sample_day << jms_ai_dw__yl_check_threesegcode_sample
